package mr

import (
	"crypto/sha256"
	"encoding/json"
	"fmt"
	"log"
	"sync"
	"time"
)

type Job struct {
	FuncName   string
	ParamsJson []byte
	CreatedAt  int64
}

// 生成的hash作为 ack 记录任务是否被响应的key
func (job *Job) Hash() [32]byte {
	jobStr := fmt.Sprintf("%s %s %d", job.FuncName, job.ParamsJson, job.CreatedAt)
	return sha256.Sum256([]byte(jobStr))
}

func (job *Job) IsEmpty() bool {
	return job.FuncName == "" || job.FuncName == "wait"
}

// 用一个map来记录没有ack的任务
// 之所以不用sync.Map，是因为那个是读多写少的时候性能好，但是目前的场景，读写差不多
// 额外提一句，这个jobAck只是维护了key是否在map中，并不关心job是啥，甚至是不是job，他关心的只有key
type jobAck struct {
	jobsDone map[[32]byte]bool
	lock     sync.Mutex
}

func NewJobAck() *jobAck {
	return &jobAck{
		jobsDone: map[[32]byte]bool{},
	}
}

func (ack *jobAck) IsAllAck() bool {
	ack.lock.Lock()
	defer ack.lock.Unlock()
	return len(ack.jobsDone) == 0
}

func (ack *jobAck) IsAck(key [32]byte) bool {
	ack.lock.Lock()
	defer ack.lock.Unlock()
	_, exit := ack.jobsDone[key]
	return !exit
}

func (ack *jobAck) Ack(key [32]byte) {
	ack.lock.Lock()
	defer ack.lock.Unlock()
	delete(ack.jobsDone, key)
}

func (ack *jobAck) WaitAck(key [32]byte) {
	ack.lock.Lock()
	defer ack.lock.Unlock()
	ack.jobsDone[key] = false
}

// 一个line作为任务队列，一个ack用来维护未响应的任务
type JobLine struct {
	line []Job
	ack  jobAck
	lock sync.Mutex
}

func NewJobLine() *JobLine {
	return &JobLine{
		ack: *NewJobAck(),
	}
}

func (jobLine *JobLine) Len() int {
	return len(jobLine.line)
}

func (jobLine *JobLine) Pop() Job {
	jobLine.lock.Lock()
	defer jobLine.lock.Unlock()
	switch len(jobLine.line) {
	case 0:
		return Job{}
	case 1:
		job := jobLine.line[0]
		jobLine.line = []Job{}
		return job
	default:
		job := jobLine.line[0]
		jobLine.line = jobLine.line[1:]
		return job
	}
}

func (jobLine *JobLine) Add(jobs ...Job) {
	jobLine.lock.Lock()
	defer jobLine.lock.Unlock()
	jobLine.line = append(jobLine.line, jobs...)
	log.Printf("get %d jobs, total %d jobs\n", len(jobs), len(jobLine.line))
}

func (jobLine *JobLine) Wait(job Job) {
	key := job.Hash()
	jobLine.ack.WaitAck(key)
	log.Printf("job: %s, %d has been in wait", job.FuncName, job.CreatedAt)
	go func() {
		time.Sleep(10 * time.Second)
		if !jobLine.ack.IsAck(key) {
			log.Printf("job: %s, %d not get ack, add it to line", job.FuncName, job.CreatedAt)
			jobLine.Add(job)
			jobLine.ack.Ack(key)
		}
	}()
}

// isEmpty and isWait
func (jobLine *JobLine) Status() (bool, bool) {
	jobLine.lock.Lock()
	defer jobLine.lock.Unlock()
	return len(jobLine.line) == 0, len(jobLine.line) == 0 && !jobLine.ack.IsAllAck()
}

func (jobLine *JobLine) PopAndWait() Job {
	job := jobLine.Pop()
	jobLine.Wait(job)
	return job
}

func (jobLine *JobLine) Ack(job Job) {
	h := job.Hash()
	log.Printf("job: %s, %d will ack", job.FuncName, job.CreatedAt)
	jobLine.ack.Ack(h)
}

func (jobLine *JobLine) IsStillWait() bool {
	return jobLine.ack.IsAllAck()
}

type MapParams struct {
	FileName string
	Contents string
}

type ReduceParams struct {
	NReduce   int
	MyReduce  int
	FileNames []string
}

func NewMapJob(fileName string, contents string) Job {
	mapParams := MapParams{fileName, contents}
	paramsJson, _ := json.Marshal(mapParams)
	return Job{"map", paramsJson, time.Now().Unix()}
}

func NewReduceJob(nreduce int, myreduce int, fileNames []string) Job {
	reduceParams := ReduceParams{NReduce: nreduce, MyReduce: myreduce, FileNames: fileNames}
	paramsJson, _ := json.Marshal(reduceParams)
	return Job{"reduce", paramsJson, time.Now().Unix()}
}

// 可能会出现任务队列为空，但是还有任务没响应的情况，这时候就要等待
func NewWaitJob() Job {
	return Job{"wait", nil, time.Now().Unix()}
}

// 通知所有的worker退出
func NewExitJob() Job {
	return Job{"exit", nil, time.Now().Unix()}
}
